package com.remoter.registry.zookeeper;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.CuratorFrameworkFactory.Builder;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent.Type;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;

import com.remoter.basic.bean.support.ConsumerModule;
import com.remoter.basic.bean.support.ProviderModule;
import com.remoter.basic.configure.IConfiguration;
import com.remoter.basic.configure.support.AbstractConfiguration;
import com.remoter.basic.registry.support.AbstractRegistry;
import com.remoter.basic.serialize.ISerialization;
import com.remoter.basic.util.Final;
import com.remoter.basic.util.ServiceLoader;
import com.remoter.basic.util.ServiceLoader.MetaName;
import com.remoter.basic.util.StringUtil;
import com.remoter.registry.zookeeper.util.FinalRegistryZookeeper;

/**
 * @author			koko
 * @date 			2017-08-08 14:56:10
 * @description 	类功能说明
 */
@MetaName("zookeeper")
public class ZookeeperRegistry extends AbstractRegistry{
	
	private CuratorFramework curatorFramework;
	private IConfiguration configuration;
	private ISerialization serialization;
	private TreeCache treeCache;
	
	public ZookeeperRegistry() {
		this.configuration = AbstractConfiguration.getConfiguration();
		this.serialization = ServiceLoader.getService(ISerialization.class,this.configuration.getOption(Final.O_VAL_SERIALIZE));
		this.start();
	}

	@Override
	public synchronized boolean attachProvider(ProviderModule providerModule){
		if(null == providerModule || this.curatorFramework.getState() != CuratorFrameworkState.STARTED){
			return false;
		}
		String key = providerModule.getKey();
		try(ByteArrayOutputStream baos = new ByteArrayOutputStream()){
			if(this.curatorFramework.checkExists().forPath(key) != null){
				this.curatorFramework.delete().forPath(key);
			}
			this.serialization.serialize(providerModule,baos,ProviderModule.class);
			this.curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(key,baos.toByteArray());
			return true;
		}catch(Exception e){
			logger.error(e.getMessage(),e);
			return false;
		}
	}

	@Override
	public synchronized boolean attachConsumer(ConsumerModule consumerModule){
		if(null == consumerModule || this.curatorFramework.getState() != CuratorFrameworkState.STARTED){
			return false;
		}
		String key = consumerModule.getKey();
		try(ByteArrayOutputStream baos = new ByteArrayOutputStream()){
			if(this.curatorFramework.checkExists().forPath(key) != null){
				this.curatorFramework.delete().forPath(key);
			}
			this.serialization.serialize(consumerModule,baos,ConsumerModule.class);
			this.curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(key,baos.toByteArray());
			return true;
		}catch(Exception e){
			logger.error(e.getMessage(),e);
			return false;
		}
	}

	@Override
	public synchronized boolean detachProvider(ProviderModule providerModule){
		if(null == providerModule || this.curatorFramework.getState() != CuratorFrameworkState.STARTED){
			return false;
		}
		String key = providerModule.getKey();
		try{
			if(this.curatorFramework.checkExists().forPath(key) != null){
				this.curatorFramework.delete().forPath(key);
			}
			return true;
		}catch(Exception e){
			logger.error(e.getMessage(),e);
			return false;
		}
	}

	@Override
	public synchronized boolean detachConsumer(ConsumerModule consumerModule) {
		if(null == consumerModule || this.curatorFramework.getState() != CuratorFrameworkState.STARTED){
			return false;
		}
		String key = consumerModule.getKey();
		try{
			if(this.curatorFramework.checkExists().forPath(key) != null){
				this.curatorFramework.delete().forPath(key);
			}
			return true;
		}catch(Exception e){
			logger.error(e.getMessage(),e);
			return false;
		}
	}
	
	@Override
	public synchronized boolean updateConsumer(ConsumerModule oldConsumerModule, ConsumerModule newConsumerModule){
		String oldKey = oldConsumerModule.getKey();
		String newKey = newConsumerModule.getKey();
		try(ByteArrayOutputStream baos = new ByteArrayOutputStream()){
			this.serialization.serialize(newConsumerModule,baos,ConsumerModule.class);
			if(this.curatorFramework.checkExists().forPath(oldKey) != null){
				this.curatorFramework.delete().forPath(oldKey);
			}
			if(this.curatorFramework.checkExists().forPath(newKey) != null){
				this.curatorFramework.delete().forPath(newKey);
			}
			this.curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(newKey,baos.toByteArray());
			return true;
		}catch(Exception e){
			e.printStackTrace();
			logger.error(e.getMessage(),e);
			return false;
		}
	}
	
	@Override
	public synchronized void close() throws IOException {
		super.close();
		CloseableUtils.closeQuietly(this.treeCache);
		CloseableUtils.closeQuietly(this.curatorFramework);
	}
	
	@Override
	public List<ProviderModule> listProviders(){
		List<ProviderModule> result = new ArrayList<ProviderModule>();
		this.list(ProviderModule.class, "/", result, Match.PROVIDER_MATCH);
		return result;
	}
	
	@Override
	public List<ConsumerModule> listConsumers() {
		List<ConsumerModule> result = new ArrayList<ConsumerModule>();
		this.list(ConsumerModule.class, "/", result, Match.CONSUMER_MATCH);
		return result;
	}

	@SuppressWarnings("unchecked")
	private <T> void list(Class<T> clazz,String path,List<T> list,Match match){
		try{
			if(null != this.curatorFramework.checkExists().forPath(path)){
				List<String> paths = this.curatorFramework.getChildren().forPath(path);
				if(null != paths && paths.size() > 0){
					for(String p : paths){
						if(!p.startsWith("/")){
							p = "/" + p;
						}
						if(path.endsWith("/")){
							path = path.substring(0,path.length()-1);
						}
						p = path + p;
						if(match.matching(p)){
							byte[] byteData = this.curatorFramework.getData().forPath(p);
							if(null != byteData && byteData.length > 0){
								try(ByteArrayInputStream bais = new ByteArrayInputStream(byteData);){
									Object obj = this.serialization.deserialize(bais,clazz);
									list.add((T)obj);
								}
							}
						}
						this.list(clazz,p,list,match);
					}
				}
			}
		}catch(Exception e){
			logger.warn(e.getMessage(),e);
		}
	}
	
	private synchronized void start(){
		Builder builder = CuratorFrameworkFactory.builder();
		builder.retryPolicy(new ExponentialBackoffRetry(configuration.getOption(FinalRegistryZookeeper.O_REGISTRY_RETRY_TIME_SLEEP),configuration.getOption(FinalRegistryZookeeper.O_REGISTRY_RETRY_TIME_COUNT)));
		builder.connectionTimeoutMs(configuration.getOption(FinalRegistryZookeeper.O_REGISTRY_CONNECTION_TIMEOUT));
		builder.connectString(configuration.getOption(FinalRegistryZookeeper.O_REGISTRY_CONNECTION_ADDRESS));
		builder.namespace(configuration.getOption(FinalRegistryZookeeper.O_REGISTRY_NAMESPACE));
		builder.sessionTimeoutMs(configuration.getOption(FinalRegistryZookeeper.O_REGISTRY_SESSION_TIMEOUT));
		builder.defaultData(null);
		this.curatorFramework = builder.build();
		this.curatorFramework.start();
		this.watching("/");
	}
	
	private void watching(String path){
		try{
			if(StringUtil.isBlank(path)){
				path = "/";
			}
			this.treeCache = new TreeCache(this.curatorFramework,path);
			treeCache.getListenable().addListener(new TreeCacheListener(){
				@Override
				public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
					ChildData data = event.getData();
					if(null != data){
						String path = data.getPath();
						boolean isProvider = Match.PROVIDER_MATCH.matching(path);
						boolean isConsuner = Match.CONSUMER_MATCH.matching(path);
						if(isProvider){
							try(ByteArrayInputStream bais = new ByteArrayInputStream(data.getData())){
								ProviderModule providerModule = (ProviderModule)ZookeeperRegistry.this.serialization.deserialize(bais,ProviderModule.class);
								if(event.getType() == Type.NODE_ADDED){
									ZookeeperRegistry.this.fireProviderCreate(path,providerModule);
								}else if(event.getType() == Type.NODE_REMOVED){
									ZookeeperRegistry.this.fireProviderDelete(path,providerModule);
								}else if(event.getType() == Type.NODE_UPDATED){
									ZookeeperRegistry.this.fireProviderUpdate(path,providerModule);
								}
							}
						}
						if(isConsuner){
							try(ByteArrayInputStream bais = new ByteArrayInputStream(data.getData())){
								ConsumerModule consumerModule = (ConsumerModule)ZookeeperRegistry.this.serialization.deserialize(bais,ConsumerModule.class);
								if(event.getType() == Type.NODE_ADDED){
									ZookeeperRegistry.this.fireConsumerCreate(path,consumerModule);
								}else if(event.getType() == Type.NODE_REMOVED){
									ZookeeperRegistry.this.fireConsumerDelete(path,consumerModule);
								}else if(event.getType() == Type.NODE_UPDATED){
									ZookeeperRegistry.this.fireConsumerUpdate(path,consumerModule);
								}
							}
						}
					}
				}
			});
			treeCache.start();
		}catch(Exception e){
			logger.error(e.getMessage());
		}
	}
	
	private static interface Match{
		public static final Match PROVIDER_MATCH = new ProviderMatch();
		public static final Match CONSUMER_MATCH = new ConsumerMatch();
		public boolean matching(String path);
	}
	
	private static class ProviderMatch implements Match{
		@Override
		public boolean matching(String path) {
			if(StringUtil.isBlank(path)){
				return false;
			}
			if(path.indexOf("/providers/")>0){
				return true;
			}else{
				return false;
			}
		}
	}
	
	private static class ConsumerMatch implements Match{
		@Override
		public boolean matching(String path) {
			if(StringUtil.isBlank(path)){
				return false;
			}
			if(path.indexOf("/consumers/")>0){
				return true;
			}else{
				return false;
			}
		}
	}
	
}